package com.yjd.msg.comm.listener;

import com.alibaba.fastjson.JSON;
import com.aliyun.mns.client.MNSClient;
import com.aliyun.mns.model.Message;
import com.yjd.comm.base.enums.DbWREnums;
import com.yjd.comm.msg.MnsMsgInfo;
import com.yjd.comm.util.FrameUtil;
import com.yjd.comm.util.StringUtil;
import com.yjd.msg.comm.mns.MessageReceiver;
import com.yjd.msg.comm.model.MsgTypeMappedMnsModel;
import com.yjd.msg.comm.service.IMsgTypeMappedMnsService;
import com.yjd.msg.comm.util.AliyunClient;
import org.apache.log4j.Logger;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import javax.annotation.Resource;
import java.util.List;
import java.util.stream.Collectors;


/**
 * 服务器启动监听阿里云消息
 * Created by yjd-wap on 2018/1/23.
 */
public class AliyunMnsListener implements ApplicationListener<ContextRefreshedEvent> {

    private static final Logger logger = Logger.getLogger(AliyunMnsListener.class);

    @Resource
    private ThreadPoolTaskExecutor executorThreadPool;//线程池

    @Resource
    private AliyunClient aliyunClient;//阿里云客户端

    @Resource
    private IMsgTypeMappedMnsService msgTypeMappedMnsService;

    /**
     * 系统启动与阿里云mns建立长连接
     * 拉取各个队列数据
     * @param contextRefreshedEvent
     */
    @Override
    public void onApplicationEvent(ContextRefreshedEvent contextRefreshedEvent) {
        if(contextRefreshedEvent.getApplicationContext().getParent() == null) {
            try {
                final MNSClient mnsClient = aliyunClient.getClient();
                //加载需要拉取的队列
                List<MsgTypeMappedMnsModel> queueList = msgTypeMappedMnsService.getModelList(FrameUtil.newHashMap("mapped_type", 1, "status", 1), DbWREnums.WRITE);
                List<String> nameList = queueList.stream().map(MsgTypeMappedMnsModel::getMns_name).distinct().collect(Collectors.toList());
                //每个队列启动一个对应线程
                for (final String name : nameList) {
                    Thread queueThread = new Thread(new Runnable() {
                        public void run() {
                            queueWorkerListener(mnsClient, name);
                        }
                    });
                    queueThread.start();
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 监听每个队列的消息
     * @param mnsClient
     * @param queueName
     */
        public void queueWorkerListener(MNSClient mnsClient, final String queueName){
            logger.info("########队列:"+queueName+ "启动监听########");
        MessageReceiver receiver = new MessageReceiver(queueName, mnsClient, queueName);
        while (true) {
            Message message = receiver.receiveMessage();
            logger.info("监听器拉取到消息:" + JSON.toJSONString(message));
            //拉取到消息准备消费
            final MnsMsgInfo mnsMsgInfo = StringUtil.getObject(message.getMessageBody(),MnsMsgInfo.class);
            final String receiptHandle = message.getReceiptHandle();
            //向线程池中添加一个任务
            executorThreadPool.execute(new Runnable() {
                public void run() {
                    MsgDisposeUtil.onMessage(mnsMsgInfo,receiptHandle,queueName); //添加一个任务
                }
            } );
        }
    }



    public ThreadPoolTaskExecutor getExecutorThreadPool() {
        return executorThreadPool;
    }

    public void setExecutorThreadPool(ThreadPoolTaskExecutor executorThreadPool) {
        this.executorThreadPool = executorThreadPool;
    }

    public AliyunClient getAliyunClient() {
        return aliyunClient;
    }

    public void setAliyunClient(AliyunClient aliyunClient) {
        this.aliyunClient = aliyunClient;
    }
}
